// Copyright (c) 2023 Huawei Device Co., Ltd.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::error::Error;
use crate::event_handler::event_queue::EventQueue;
use crate::event_handler::inner_event::Closure::{ActorTask, Task};
use crate::event_handler::inner_event::DefaultType;
use crate::event_handler::{EventRunnerHandle, InnerEvent, Priority};
use crate::DispatcherHandle;
use std::cell::{RefCell, UnsafeCell};
use std::future::Future;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::sync::oneshot;
use tokio::task::JoinHandle;

pub(crate) type SharedHandler = UnsafeCell<dyn DistributeEvent + Send + Sync>;

thread_local! {
    static HANDLER: RefCell<Option<EventHandler>> = RefCell::new(None);
}

/// Implementing this trait for a user struct makes the struct an `EventHandler`. An `EventHandler`
/// is guaranteed to process events serially.
pub trait ToEventHandler: Sized + Send + Sync + 'static {
    /// Defines a function which will be called every time a `InnerEvent` received.
    fn process_event(&mut self, _event: &InnerEvent<DefaultType>) {}
}

/// Auto implemented for which implemented `ToEventHandler`.
pub trait ToEventHandlerExt: EventHandlerSealed + Sized {
    /// Turns self into an `EventHandler`.
    fn to_event_handler(self, runner: &EventRunnerHandle) -> EventHandler;
}

impl<T: ToEventHandler> EventHandlerSealed for T {}

impl<T: ToEventHandler> ToEventHandlerExt for T {
    /// Starts `EventHandler` with a specific `EventRunner`. See [`EventRunner`].
    ///
    /// [`EventRunner`]: crate::event_handler::EventRunner
    fn to_event_handler(self, runner: &EventRunnerHandle) -> EventHandler {
        EventHandler {
            handler: Arc::new(UnsafeCell::new(self)),
            runner: runner.clone(),
        }
    }
}

/// Implementing this trait for a user struct makes the struct an `Actor`
pub trait ToActor: Sized + Send + 'static {}

/// Auto implemented for which implemented `ToActor`.
pub trait ToActorExt: ActorSealed + Sized {
    /// Turns self into an `Actor` and gets `ActorRef`.
    fn start_actor_on(self, dispacher: &DispatcherHandle) -> ActorRef<Self>;
}

impl<T: ToActor> ActorSealed for T {}

impl<T: ToActor> ToActorExt for T {
    /// Starts an actor on specific dispatcher. It is used to limit actor running on the same thread.
    fn start_actor_on(self, dispacher: &DispatcherHandle) -> ActorRef<Self> {
        let queue = Arc::new(EventQueue::new());
        dispacher.spawn(run(self, queue.clone()));
        ActorRef { queue }
    }
}

pub(crate) trait DistributeEvent {
    fn distribute(&mut self, event: InnerEvent<DefaultType>);
}

impl<T: ToEventHandler> DistributeEvent for T {
    fn distribute(&mut self, mut event: InnerEvent<DefaultType>) {
        match event.task_callback() {
            None => self.process_event(&event),
            Some(Task(f)) => f(),
            Some(ActorTask(_)) => panic!("Invalid event"),
        }
    }
}

async fn run<T>(mut actor: T, rx: Arc<EventQueue<T>>) {
    while let Some(mut event) = rx.get().await {
        match event.task_callback() {
            None => panic!("Invalid event"),
            Some(Task(_)) => panic!("Invalid event"),
            Some(ActorTask(f)) => f(&mut actor),
        }
    }
}

/// An event handler which is guaranteed to process events serially.
#[derive(Clone)]
pub struct EventHandler {
    handler: Arc<SharedHandler>,
    runner: EventRunnerHandle,
}

impl EventHandler {
    pub(crate) fn new(handler: Arc<SharedHandler>, runner: EventRunnerHandle) -> Self {
        Self { handler, runner }
    }

    pub(crate) fn distribute(&self, event: InnerEvent<DefaultType>) {
        HANDLER.with(|h| *h.borrow_mut() = Some(self.clone()));
        // SAFETY: The method `distribute` is called serially for each
        // `EventHandler`, and is the only way to access the `SharedHandler`,
        // so it's safe to get the mutable reference of `SharedHandler`.
        let handler = unsafe { &mut *self.handler.get() };
        handler.distribute(event);
        HANDLER.with(|h| *h.borrow_mut() = None);
    }

    /// Sends an event to the `EventHandler`. `delay_time` means to process this event after the
    /// specified number of milliseconds.
    pub fn send_event(
        &self,
        mut event: InnerEvent<DefaultType>,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error> {
        let now = SystemTime::now();
        event
            .set_send_time(now)
            .set_handle_time(
                now.checked_add(Duration::from_millis(delay_time))
                    .ok_or(Error::InvalidTime)?,
            )
            .set_owner(&self.handler);
        self.runner.put(event, priority)?;
        Ok(())
    }

    /// Sends a task to the `EventHandler`. `delay_time` means to process this event after the
    /// specified number of milliseconds.
    pub fn post_task<F>(
        &self,
        task: F,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error>
    where
        F: FnOnce() + Send + 'static,
    {
        let event = InnerEvent::new_task(name, task);
        self.send_event(event, delay_time, priority)
    }

    /// Sends an event to the `EventHandler` and waits for the event being processed to end.
    /// `delay_time` means to process this event after the specified number of milliseconds.
    pub async fn send_sync_event(
        &self,
        mut event: InnerEvent<DefaultType>,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error> {
        if matches!(priority, Priority::Idle) {
            return Err(Error::InvalidPriority);
        }

        let waiter = event.create_waiter();
        self.send_event(event, delay_time, priority)?;
        // Wait until event is processed.
        waiter.wait().await;
        Ok(())
    }

    /// Sends a task to the `EventHandler` and waits for the task executing to end. `delay_time`
    /// means to process this event after the specified number of milliseconds.
    pub async fn post_sync_task<F>(
        &self,
        task: F,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error>
    where
        F: FnOnce() + Send + 'static,
    {
        let event = InnerEvent::new_task(name, task);
        self.send_sync_event(event, delay_time, priority).await
    }

    /// Gets the current `EventHandler`'s reference, should only be called within the event
    /// processing context, which means [`ToEventHandler::process_event`] or task closure sent to
    /// the `EventHandler`.
    pub fn current() -> Option<EventHandler> {
        HANDLER.with(|h| h.borrow().clone())
    }

    /// Removes the sent event by event id.
    pub fn remove_event(&self, _id: u32) {
        todo!()
    }

    /// Removes the sent task by name.
    pub fn remove_task(&self, _name: &str) {
        todo!()
    }

    /// Removes all the sent event and task of the `EventHandler`.
    pub fn remove_all(&self) {
        todo!()
    }

    /// Gets the `EventRunnerHandle` of the `EventHandler`.
    pub fn event_runner(&self) -> &EventRunnerHandle {
        &self.runner
    }

    /// Spawns a task to the `EventRunner` of the `EventHandler`.
    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.runner.spawn(future)
    }
}

/// Implements the trait for user actor struct with a specific msg type. It can be implemented multi
/// times for the same actor with different msg types.
pub trait Handler<T: Send + 'static>: 'static {
    /// Return type of `handle`.
    type ReturnType: Send;

    /// Function to handle the msg.
    fn handle(&mut self, msg: T) -> Self::ReturnType;
}

pub struct ActorRef<A> {
    queue: Arc<EventQueue<A>>,
}

impl<A> ActorRef<A> {
    /// Sends a msg to the actor. `delay_time` means to process this event after the
    /// specified number of milliseconds.
    /// User should implement `Handler` trait for the msg type first. See [`Handler`].
    ///
    /// [`Handler`]: Handler
    pub fn send_msg<T>(
        &self,
        msg: T,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error>
    where
        A: Handler<T> + Send,
        T: Send + 'static,
    {
        let event = InnerEvent::new_actor_task(name, move |actor| {
            let _ = A::handle(actor, msg);
        });
        self.send_event(event, delay_time, priority)
    }

    /// Sends an event to the actor. `delay_time` means to process this event after the
    /// specified number of milliseconds.
    fn send_event(
        &self,
        mut event: InnerEvent<A>,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error> {
        let now = SystemTime::now();
        event.set_send_time(now).set_handle_time(
            now.checked_add(Duration::from_millis(delay_time))
                .ok_or(Error::InvalidTime)?,
        );
        self.queue.put(event, priority);
        Ok(())
    }

    /// Sends a task to the actor. `delay_time` means to process this event after the
    /// specified number of milliseconds.
    pub fn send_task<F>(
        &self,
        task: F,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error>
    where
        F: FnOnce(&mut A) + Send + 'static,
    {
        let event = InnerEvent::new_actor_task(name, task);
        self.send_event(event, delay_time, priority)
    }

    /// Sends a request(which is a msg) to the actor, and waits for the response(`Handler::ReturnType`).
    /// `delay_time` means to process this event after the specified number of milliseconds.
    pub async fn send_sync_msg<T>(
        &self,
        msg: T,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<<A as Handler<T>>::ReturnType, Error>
    where
        A: Handler<T> + Send,
        T: Send + 'static,
    {
        let (tx, rx) = oneshot::channel();
        let event = InnerEvent::new_actor_task(name, move |actor| {
            let res = A::handle(actor, msg);
            let _ = tx.send(res);
        });
        self.send_event(event, delay_time, priority)?;
        rx.await.map_err(|e| Error::Recv(e.to_string()))
    }

    /// Sends an event to the actor and waits for the event being processed to end.
    /// `delay_time` means to process this event after the specified number of milliseconds.
    async fn send_sync_event(
        &self,
        mut event: InnerEvent<A>,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error> {
        if matches!(priority, Priority::Idle) {
            return Err(Error::InvalidPriority);
        }

        let waiter = event.create_waiter();
        self.send_event(event, delay_time, priority)?;
        // Wait until event is processed.
        waiter.wait().await;
        Ok(())
    }

    /// Sends a task to the actor and waits for the task executing to end. `delay_time`
    /// means to process this event after the specified number of milliseconds.
    pub async fn send_sync_task<F>(
        &self,
        task: F,
        name: &str,
        delay_time: u64,
        priority: Priority,
    ) -> Result<(), Error>
    where
        F: FnOnce(&mut A) + Send + 'static,
    {
        let event = InnerEvent::new_actor_task(name, task);
        self.send_sync_event(event, delay_time, priority).await
    }

    /// Stops the actor running referred by the `ActorRef`. All
    /// the events haven't been processed will be dropped.
    pub fn stop(&self) {
        self.queue.finish();
    }
}

pub trait EventHandlerSealed {}

pub trait ActorSealed {}

#[cfg(test)]
mod ut_handler {
    use crate::error::Error;
    use crate::event_handler::{
        DefaultType, EventHandler, EventRunner, InnerEvent, Priority, ToEventHandler,
        ToEventHandlerExt,
    };
    use std::sync::atomic::{AtomicU32, Ordering};
    use tokio::runtime::Runtime;

    static FLAG: AtomicU32 = AtomicU32::new(0);

    struct TestEventHandler;

    impl ToEventHandler for TestEventHandler {
        fn process_event(&mut self, event: &InnerEvent<DefaultType>) {
            FLAG.fetch_add(event.id(), Ordering::Relaxed);
        }
    }

    /// Make sure all the tests of this mod run in one thread.
    #[test]
    fn ut_all_tests() {
        let runtime = Runtime::new().unwrap();
        let runner = EventRunner::create_detached("test").unwrap();
        let event_handler = TestEventHandler.to_event_handler(&runner);

        ut_distribute(&event_handler);
        runtime.block_on(ut_send_event(&event_handler));
        runtime.block_on(ut_send_sync_event(&event_handler));
        runtime.block_on(ut_post_sync_task(&event_handler));
        runner.stop();
    }

    /// UT test for `EventHandler::distribute`.
    ///
    /// # Brief
    /// 1. Create events using `InnerEvent::new` and `InnerEvent::new_task` methods.
    /// 2. Call `EventHandler::distribute` with these events, and check the results.
    fn ut_distribute(event_handler: &EventHandler) {
        let event = InnerEvent::new(10);
        event_handler.distribute(event);
        assert_eq!(FLAG.load(Ordering::Relaxed), 10);

        let event = InnerEvent::new_task("task", || {
            FLAG.fetch_sub(10, Ordering::Relaxed);
        });
        event_handler.distribute(event);
        assert_eq!(FLAG.load(Ordering::Relaxed), 0);
    }

    /// UT test for `EventHandler::send_event`.
    ///
    /// # Brief
    /// 1. Test different valid delay_time and priority.
    /// 2. Wait for all events getting processed over, and check the result.
    async fn ut_send_event(event_handler: &EventHandler) {
        // make sure idle event sent before other events.
        let mut event = InnerEvent::new(10);
        let idle_waiter = event.create_waiter();
        event_handler.send_event(event, 0, Priority::Idle).unwrap();

        let event = InnerEvent::new(10);
        event_handler
            .send_event(event, 10, Priority::Immediate)
            .unwrap();

        let event = InnerEvent::new(10);
        event_handler.send_event(event, 20, Priority::High).unwrap();

        let mut event = InnerEvent::new(10);
        let waiter = event.create_waiter();
        event_handler.send_event(event, 30, Priority::Low).unwrap();
        // wait for idle event and other events being processed.
        waiter.wait().await;
        idle_waiter.wait().await;

        assert_eq!(FLAG.load(Ordering::Relaxed), 40);
        FLAG.store(0, Ordering::Relaxed);
    }

    /// UT test for `EventHandler::send_sync_event`.
    ///
    /// # Brief
    /// 1. Test invalid `priority`.
    /// 2. Test different valid delay_time and priority.
    /// 3. Check the result of every processed event.
    async fn ut_send_sync_event(event_handler: &EventHandler) {
        let event = InnerEvent::new(10);
        let err = event_handler
            .send_sync_event(event, 0, Priority::Idle)
            .await
            .unwrap_err();
        assert!(matches!(err, Error::InvalidPriority));

        let event = InnerEvent::new(10);
        event_handler
            .send_sync_event(event, 30, Priority::Low)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 10);

        let event = InnerEvent::new(10);
        event_handler
            .send_sync_event(event, 20, Priority::Immediate)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 20);

        let event = InnerEvent::new(10);
        event_handler
            .send_sync_event(event, 10, Priority::High)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 30);

        FLAG.store(0, Ordering::Relaxed);
    }

    /// UT test for `EventHandler::post_sync_task`.
    ///
    /// # Brief
    /// 1. Test invalid `priority`.
    /// 2. Test different valid delay_time and priority.
    /// 3. Check the result of every processed task.
    async fn ut_post_sync_task(event_handler: &EventHandler) {
        let err = event_handler
            .post_sync_task(|| {}, "", 0, Priority::Idle)
            .await
            .unwrap_err();
        assert!(matches!(err, Error::InvalidPriority));

        let closure = || {
            FLAG.fetch_add(10, Ordering::Relaxed);
        };

        event_handler
            .post_task(closure, "", 0, Priority::High)
            .unwrap();

        event_handler
            .post_sync_task(closure, "", 30, Priority::Low)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 20);

        event_handler
            .post_sync_task(closure, "", 20, Priority::Immediate)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 30);

        event_handler
            .post_sync_task(closure, "", 10, Priority::High)
            .await
            .unwrap();
        assert_eq!(FLAG.load(Ordering::Relaxed), 40);

        FLAG.store(0, Ordering::Relaxed);
    }
}
